/* -------------------------------------------------------------------------
 *
 * decode.c
 *		该模块使用 xlogreader.h 的 API 解码 WAL 记录，
 *		用于逻辑解码，通过将信息传递给 reorderbuffer 模块（包含实际更改）
 *		以及 snapbuild 模块以构建合适的目录快照（以便能够
 *		在 reorderbuffer 中正确解码更改）。
 *
 * NOTE:
 *		这基本上尝试处理 reorderbuffer.c 和 snapbuild.c 的所有低级 xlog
 *		内容。有一些轻微的泄漏，特定记录的结构被用于传递数据，
 *		但这些只偶然包含了适量的数据，并且格式便利。
 *		这里没有，也不应该有关于记录内容的过多智能，
 *		除了将它们转变为更易用的格式。
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *	  src/backend/replication/logical/decode.c
 *
 * -------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/heapam.h"
#include "access/heapam_xlog.h"
#include "access/transam.h"
#include "access/xact.h"
#include "access/xlog_internal.h"
#include "access/xlogreader.h"
#include "access/xlogrecord.h"
#include "access/xlogutils.h"
#include "catalog/pg_control.h"
#include "replication/decode.h"
#include "replication/logical.h"
#include "replication/message.h"
#include "replication/origin.h"
#include "replication/reorderbuffer.h"
#include "replication/snapbuild.h"
#include "storage/standby.h"
#ifdef FDD
#include "storage/procarray.h"
#include "replication/syncrep.h"
#endif

/* 个人记录（组）的处理程序 */
static void fc_DecodeInsert(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf);
static void fc_DecodeUpdate(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf);
static void fc_DecodeDelete(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf);
static void fc_DecodeTruncate(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf);
static void fc_DecodeMultiInsert(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf);
static void fc_DecodeSpecConfirm(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf);

static void fc_DecodeCommit(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf,
						 xl_xact_parsed_commit *fc_parsed, TransactionId fc_xid,
						 bool fc_two_phase);
static void fc_DecodeAbort(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf,
						xl_xact_parsed_abort *fc_parsed, TransactionId fc_xid,
						bool fc_two_phase);
static void fc_DecodePrepare(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf,
						  xl_xact_parsed_prepare *fc_parsed);


/* 解码元组的公共函数 */
static void fc_DecodeXLogTuple(char *fc_data, Size fc_len, ReorderBufferTupleBuf *fc_tup);

/* 解码事务的辅助函数 */
static inline bool fc_FilterPrepare(LogicalDecodingContext *fc_ctx,
								 TransactionId fc_xid, const char *fc_gid);
static bool fc_DecodeTXNNeedSkip(LogicalDecodingContext *fc_ctx,
							  XLogRecordBuffer *fc_buf, Oid fc_dbId,
#ifdef FDDQ
							  RepOriginId fc_origin_id,uint32		fc_xinfo);
#else
							  RepOriginId fc_origin_id);
#endif


/*
 * 处理每个 XLogReadRecord() 的记录，并执行所需的操作以
 * 使用在逻辑解码上下文中已经设置的输出插件解码它。
 *
 * 注意：每个记录的 xid 需要通过 reorderbuffer 处理
 * （记录内容中的 xids 对此规则不相关）。
 * 这意味着对于那些本来不会经过 reorderbuffer 的记录，
 * 必须调用 ReorderBufferProcessXid()。我们不想
 * 默认为每种记录类型调用 ReorderBufferProcessXid，因为
 * 例如，如果没有它们的前一个状态，空事务可以更有效地处理。
 *
 * 我们还支持快速向前跳过记录，完全跳过某些
 * 记录类型 - 详细信息见各个记录类型。
 */
void LogicalDecodingProcessRecord(LogicalDecodingContext *fc_ctx, XLogReaderState *fc_record)
{
	XLogRecordBuffer fc_buf;
	TransactionId fc_txid;
	RmgrData	fc_rmgr;

	fc_buf.origptr = fc_ctx->reader->ReadRecPtr;
	fc_buf.endptr = fc_ctx->reader->EndRecPtr;
	fc_buf.record = fc_record;

	fc_txid = XLogRecGetTopXid(fc_record);

	/*
	 * 如果顶级 xid 有效，我们需要将子事务分配给
	 * 顶级事务。我们需要对所有记录执行此操作，因此我们
	 * 在切换之前执行它。
	 */
	if (TransactionIdIsValid(fc_txid))
	{
		ReorderBufferAssignChild(fc_ctx->reorder,
								 fc_txid,
								 XLogRecGetXid(fc_record),
								 fc_buf.origptr);
	}

	fc_rmgr = GetRmgr(XLogRecGetRmid(fc_record));

	if (fc_rmgr.rm_decode != NULL)
		fc_rmgr.rm_decode(fc_ctx, &fc_buf);
	else
	{
		/* 仅处理 xid，完成 */
		ReorderBufferProcessXid(fc_ctx->reorder, XLogRecGetXid(fc_record),
								fc_buf.origptr);
	}
}

/*
 * 处理 rmgr XLOG_ID 记录以解码记录进入 reorderBuffer()。
 */
void xlog_decode(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf)
{
	SnapBuild  *fc_builder = fc_ctx->snapshot_builder;
	uint8		fc_info = XLogRecGetInfo(fc_buf->record) & ~XLR_INFO_MASK;

	ReorderBufferProcessXid(fc_ctx->reorder, XLogRecGetXid(fc_buf->record),
							fc_buf->origptr);

	switch (fc_info)
	{
			/* 此处也用于 END_OF_RECOVERY 检查点 */
		case XLOG_CHECKPOINT_SHUTDOWN:
		case XLOG_END_OF_RECOVERY:
			SnapBuildSerializationPoint(fc_builder, fc_buf->origptr);

			break;
		case XLOG_CHECKPOINT_ONLINE:

			/*
			 * 一个 RUNNING_XACTS 记录已在此附近记录，我们
			 * 可以从那里重新启动。
			 */
			break;
		case XLOG_NOOP:
		case XLOG_NEXTOID:
		case XLOG_SWITCH:
		case XLOG_BACKUP_END:
		case XLOG_PARAMETER_CHANGE:
		case XLOG_RESTORE_POINT:
		case XLOG_FPW_CHANGE:
		case XLOG_FPI_FOR_HINT:
		case XLOG_FPI:
		case XLOG_OVERWRITE_CONTRECORD:
			break;
		default:
			elog(ERROR, "unexpected RM_XLOG_ID record type: %u", fc_info);
	}
}

/*
 * 处理 rmgr XACT_ID 记录以解码记录进入 reorderBuffer()。
 */
void xact_decode(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf)
{
	SnapBuild  *fc_builder = fc_ctx->snapshot_builder;
	ReorderBuffer *fc_reorder = fc_ctx->reorder;
	XLogReaderState *fc_r = fc_buf->record;
	uint8		fc_info = XLogRecGetInfo(fc_r) & XLOG_XACT_OPMASK;

	/*
	 * 如果快照尚未完全构建，我们无法解码任何内容，因此
	 * 退出。
	 */
	if (SnapBuildCurrentState(fc_builder) < SNAPBUILD_FULL_SNAPSHOT)
		return;

	switch (fc_info)
	{
		case XLOG_XACT_COMMIT:
		case XLOG_XACT_COMMIT_PREPARED:
			{
				xl_xact_commit *fc_xlrec;
				xl_xact_parsed_commit fc_parsed;
				TransactionId fc_xid;
				bool		fc_two_phase = false;

				fc_xlrec = (xl_xact_commit *) XLogRecGetData(fc_r);
				ParseCommitRecord(XLogRecGetInfo(fc_buf->record), fc_xlrec, &fc_parsed);

				if (!TransactionIdIsValid(fc_parsed.twophase_xid))
					fc_xid = XLogRecGetXid(fc_r);
				else
					fc_xid = fc_parsed.twophase_xid;

				/*
				 * 如果输出插件支持二阶段提交且
				 * 在准备时未过滤事务，则我们希望以两阶段
				 * 方式处理事务。
				 */
				if (fc_info == XLOG_XACT_COMMIT_PREPARED)
					fc_two_phase = !(fc_FilterPrepare(fc_ctx, fc_xid,
												fc_parsed.twophase_gid));

				fc_DecodeCommit(fc_ctx, fc_buf, &fc_parsed, fc_xid, fc_two_phase);
				break;
			}
		case XLOG_XACT_ABORT:
		case XLOG_XACT_ABORT_PREPARED:
			{
				xl_xact_abort *fc_xlrec;
				xl_xact_parsed_abort fc_parsed;
				TransactionId fc_xid;
				bool		fc_two_phase = false;

				fc_xlrec = (xl_xact_abort *) XLogRecGetData(fc_r);
				ParseAbortRecord(XLogRecGetInfo(fc_buf->record), fc_xlrec, &fc_parsed);

				if (!TransactionIdIsValid(fc_parsed.twophase_xid))
					fc_xid = XLogRecGetXid(fc_r);
				else
					fc_xid = fc_parsed.twophase_xid;

				/*
				 * 如果输出插件支持二阶段提交且
				 * 在准备时未过滤事务，则我们希望以两阶段
				 * 方式处理事务。
				 */
				if (fc_info == XLOG_XACT_ABORT_PREPARED)
					fc_two_phase = !(fc_FilterPrepare(fc_ctx, fc_xid,
												fc_parsed.twophase_gid));

				fc_DecodeAbort(fc_ctx, fc_buf, &fc_parsed, fc_xid, fc_two_phase);
				break;
			}
		case XLOG_XACT_ASSIGNMENT:

			/*
			 * 在处理每个记录时，如果需要，我们将子事务分配给
			 * 顶级事务。因此，我们这里不需要做任何事情。参见
			 * LogicalDecodingProcessRecord。
			 */
			break;
		case XLOG_XACT_INVALIDATIONS:
			{
				TransactionId fc_xid;
				xl_xact_invals *fc_invals;

				fc_xid = XLogRecGetXid(fc_r);
				fc_invals = (xl_xact_invals *) XLogRecGetData(fc_r);

				/*
				 * 执行无 xid 事务的无效化，
				 * 否则，累积它们，以便可以在
				 * 提交时处理。
				 */
				if (TransactionIdIsValid(fc_xid))
				{
					if (!fc_ctx->fast_forward)
						ReorderBufferAddInvalidations(fc_reorder, fc_xid,
													  fc_buf->origptr,
													  fc_invals->nmsgs,
													  fc_invals->msgs);
					ReorderBufferXidSetCatalogChanges(fc_ctx->reorder, fc_xid,
													  fc_buf->origptr);
				}
				else if ((!fc_ctx->fast_forward))
					ReorderBufferImmediateInvalidation(fc_ctx->reorder,
													   fc_invals->nmsgs,
													   fc_invals->msgs);
			}
			break;
		case XLOG_XACT_PREPARE:
			{
				xl_xact_parsed_prepare fc_parsed;
				xl_xact_prepare *fc_xlrec;

				/* 好的，解析它 */
				fc_xlrec = (xl_xact_prepare *) XLogRecGetData(fc_r);
				ParsePrepareRecord(XLogRecGetInfo(fc_buf->record),
								   fc_xlrec, &fc_parsed);

				/*
				 * 如果输出插件支持二阶段提交且
				 * 在准备时未过滤事务，则我们希望以两阶段
				 * 方式处理事务。
				 */
				if (fc_FilterPrepare(fc_ctx, fc_parsed.twophase_xid,
								  fc_parsed.twophase_gid))
				{
					ReorderBufferProcessXid(fc_reorder, fc_parsed.twophase_xid,
											fc_buf->origptr);
					break;
				}

				/*
				 * 请注意，如果准备的事务独占锁定了 [用户]
				 * 目录表，则解码准备可能会阻塞，
				 * 直到主事务提交，因为它需要锁定
				 * 目录表。
				 *
				 * XXX 现在，如果准备事务在等待
				 * 将其逻辑复制以进行分布式 2PC，则可能导致死锁。
				 * 可以通过禁止独占锁定 [用户] 目录表的
				 * 准备事务来避免这种情况，但目前，我们请用户
				 * 不要执行这样的操作。
				 */
				fc_DecodePrepare(fc_ctx, fc_buf, &fc_parsed);
				break;
			}
		default:
			elog(ERROR, "unexpected RM_XACT_ID record type: %u", fc_info);
	}
}

/*
 * 处理 rmgr STANDBY_ID 记录以解码记录进入 reorderBuffer()。
 */
void standby_decode(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf)
{
	SnapBuild  *fc_builder = fc_ctx->snapshot_builder;
	XLogReaderState *fc_r = fc_buf->record;
	uint8		fc_info = XLogRecGetInfo(fc_r) & ~XLR_INFO_MASK;

	ReorderBufferProcessXid(fc_ctx->reorder, XLogRecGetXid(fc_r), fc_buf->origptr);

	switch (fc_info)
	{
		case XLOG_RUNNING_XACTS:
			{
				xl_running_xacts *fc_running = (xl_running_xacts *) XLogRecGetData(fc_r);

				SnapBuildProcessRunningXacts(fc_builder, fc_buf->origptr, fc_running);

				/*
				 * 中止所有我们跟踪的旧于记录的
				 * oldestRunningXid 的事务。这是执行此操作的最
				 * 方便的地方，因为与关机或恢复结束检查点
				 * 相比，我们拥有关于所有正在运行的事务的信息，
				 * 包括准备好的事务，而关机检查点只知道没有
				 * 非准备事务正在进行。
				 */
				ReorderBufferAbortOld(fc_ctx->reorder, fc_running->oldestRunningXid);
			}
			break;
		case XLOG_STANDBY_LOCK:
			break;
		case XLOG_INVALIDATIONS:

			/*
			 * 我们通过 XLOG_XACT_INVALIDATIONS 在命令级别处理无效化。
			 * 所以我们这里不需要做任何事情。
			 */
			break;
		default:
			elog(ERROR, "unexpected RM_STANDBY_ID record type: %u", fc_info);
	}
}

/*
 * 处理 rmgr HEAP2_ID 记录以解码记录进入 reorderBuffer()。
 */
void heap2_decode(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf)
{
	uint8		fc_info = XLogRecGetInfo(fc_buf->record) & XLOG_HEAP_OPMASK;
	TransactionId fc_xid = XLogRecGetXid(fc_buf->record);
	SnapBuild  *fc_builder = fc_ctx->snapshot_builder;

	ReorderBufferProcessXid(fc_ctx->reorder, fc_xid, fc_buf->origptr);

	/*
	 * 如果我们没有快照或只是快速向前推进，那么就没有
	 * 解码更改的意义。
	 */
	if (SnapBuildCurrentState(fc_builder) < SNAPBUILD_FULL_SNAPSHOT ||
		fc_ctx->fast_forward)
		return;

	switch (fc_info)
	{
		case XLOG_HEAP2_MULTI_INSERT:
			if (!fc_ctx->fast_forward &&
				SnapBuildProcessChange(fc_builder, fc_xid, fc_buf->origptr))
				fc_DecodeMultiInsert(fc_ctx, fc_buf);
			break;
		case XLOG_HEAP2_NEW_CID:
			{
				xl_heap_new_cid *fc_xlrec;

				fc_xlrec = (xl_heap_new_cid *) XLogRecGetData(fc_buf->record);
				SnapBuildProcessNewCid(fc_builder, fc_xid, fc_buf->origptr, fc_xlrec);

				break;
			}
		case XLOG_HEAP2_REWRITE:

			/*
			 * 尽管这些记录仅存在以满足逻辑
			 * 解码的需要，但所有工作都作为崩溃或归档
			 * 恢复的一部分发生，因此我们这里不需要做任何事情。
			 */
			break;

			/*
			 * 此处的所有其他内容都是我们不感兴趣的低级物理内容。
			 */
		case XLOG_HEAP2_FREEZE_PAGE:
		case XLOG_HEAP2_PRUNE:
		case XLOG_HEAP2_VACUUM:
		case XLOG_HEAP2_VISIBLE:
		case XLOG_HEAP2_LOCK_UPDATED:
			break;
		default:
			elog(ERROR, "unexpected RM_HEAP2_ID record type: %u", fc_info);
	}
}

/*
 * 处理 rmgr HEAP_ID 记录以解码记录进入 reorderBuffer()。
 */
void heap_decode(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf)
{
	uint8		fc_info = XLogRecGetInfo(fc_buf->record) & XLOG_HEAP_OPMASK;
	TransactionId fc_xid = XLogRecGetXid(fc_buf->record);
	SnapBuild  *fc_builder = fc_ctx->snapshot_builder;

	ReorderBufferProcessXid(fc_ctx->reorder, fc_xid, fc_buf->origptr);

	/*
	 * 如果我们没有快照或只是快速向前推进，那么就没有
	 * 解码数据更改的意义。
	 */
	if (SnapBuildCurrentState(fc_builder) < SNAPBUILD_FULL_SNAPSHOT ||
		fc_ctx->fast_forward)
		return;

	switch (fc_info)
	{
		case XLOG_HEAP_INSERT:
			if (SnapBuildProcessChange(fc_builder, fc_xid, fc_buf->origptr))
				fc_DecodeInsert(fc_ctx, fc_buf);
			break;

			/*
			 * 将 HOT 更新视为普通更新。实际上，我们可以在本地将其作为 HOT 更新
			 * 处理，而 WAL 布局也兼容，这没有什么有用的信息。
			 */
		case XLOG_HEAP_HOT_UPDATE:
		case XLOG_HEAP_UPDATE:
			if (SnapBuildProcessChange(fc_builder, fc_xid, fc_buf->origptr))
				fc_DecodeUpdate(fc_ctx, fc_buf);
			break;

		case XLOG_HEAP_DELETE:
			if (SnapBuildProcessChange(fc_builder, fc_xid, fc_buf->origptr))
				fc_DecodeDelete(fc_ctx, fc_buf);
			break;

		case XLOG_HEAP_TRUNCATE:
			if (SnapBuildProcessChange(fc_builder, fc_xid, fc_buf->origptr))
				fc_DecodeTruncate(fc_ctx, fc_buf);
			break;

		case XLOG_HEAP_INPLACE:

			/*
			 * 就定义而言，原地更新仅在目录元组上执行，
			 * 并且不能更改元组的可见性。由于我们不解码目录元组，
			 * 因此我们对记录的内容不感兴趣。
			 *
			 * 原地更新可以由带有 XID 的事务（例如，在 CREATE INDEX CONCURRENTLY 中）
			 * 或者无 XID 的事务（例如 VACUUM）使用。在前一种情况下，提交
			 * 记录将包括缓存失效，因此我们在此将事务标记为目录修改。当前这
			 * 是多余的，因为提交也将这样做，但一旦我们支持解码进行中的关系，
			 * 这将变得重要。
			 */
			if (!TransactionIdIsValid(fc_xid))
				break;

			(void) SnapBuildProcessChange(fc_builder, fc_xid, fc_buf->origptr);
			ReorderBufferXidSetCatalogChanges(fc_ctx->reorder, fc_xid, fc_buf->origptr);
			break;

		case XLOG_HEAP_CONFIRM:
			if (SnapBuildProcessChange(fc_builder, fc_xid, fc_buf->origptr))
				fc_DecodeSpecConfirm(fc_ctx, fc_buf);
			break;

		case XLOG_HEAP_LOCK:
			/* 我们暂时不关心行级锁 */
			break;

		default:
			elog(ERROR, "unexpected RM_HEAP_ID record type: %u", fc_info);
			break;
	}
}

/*
 * 询问输出插件我们是否希望跳过此 PREPARE，并稍后将
 * 此事务作为常规提交发送。
 */
static inline bool fc_FilterPrepare(LogicalDecodingContext *fc_ctx, TransactionId fc_xid,
			  const char *fc_gid)
{
	/*
	 * 如果在 PREPARE 时间没有启用二阶段事务的解码，则跳过。
	 * 在这种情况下，所有二阶段事务将被视为过滤掉，并将在
	 * COMMIT PREPARED 时作为常规事务应用。
	 */
	if (!fc_ctx->twophase)
		return true;

	/*
	 * filter_prepare 回调是可选的。如果未提供，所有
	 * 准备好的交易应当通过。
	 */
	if (fc_ctx->callbacks.filter_prepare_cb == NULL)
		return false;

	return filter_prepare_cb_wrapper(fc_ctx, fc_xid, fc_gid);
}

static inline bool fc_FilterByOrigin(LogicalDecodingContext *fc_ctx, RepOriginId fc_origin_id)
{
	if (fc_ctx->callbacks.filter_by_origin_cb == NULL)
		return false;

	return filter_by_origin_cb_wrapper(fc_ctx, fc_origin_id);
}

/*
 * 处理 rmgr LOGICALMSG_ID 记录以解码记录到重排序缓冲区。
 */
void logicalmsg_decode(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf)
{
	SnapBuild  *fc_builder = fc_ctx->snapshot_builder;
	XLogReaderState *fc_r = fc_buf->record;
	TransactionId fc_xid = XLogRecGetXid(fc_r);
	uint8		fc_info = XLogRecGetInfo(fc_r) & ~XLR_INFO_MASK;
	RepOriginId fc_origin_id = XLogRecGetOrigin(fc_r);
	Snapshot	fc_snapshot = NULL;
	xl_logical_message *fc_message;

	if (fc_info != XLOG_LOGICAL_MESSAGE)
		elog(ERROR, "unexpected RM_LOGICALMSG_ID record type: %u", fc_info);

	ReorderBufferProcessXid(fc_ctx->reorder, XLogRecGetXid(fc_r), fc_buf->origptr);

	/*
	 * 如果我们没有快照或者我们只是快速前进，那么解码消息是没有意义的。
	 */
	if (SnapBuildCurrentState(fc_builder) < SNAPBUILD_FULL_SNAPSHOT ||
		fc_ctx->fast_forward)
		return;

	fc_message = (xl_logical_message *) XLogRecGetData(fc_r);

	if (fc_message->dbId != fc_ctx->slot->data.database ||
		fc_FilterByOrigin(fc_ctx, fc_origin_id))
		return;

	if (fc_message->transactional &&
		!SnapBuildProcessChange(fc_builder, fc_xid, fc_buf->origptr))
		return;
	else if (!fc_message->transactional &&
			 (SnapBuildCurrentState(fc_builder) != SNAPBUILD_CONSISTENT ||
			  SnapBuildXactNeedsSkip(fc_builder, fc_buf->origptr)))
		return;

	/*
	 * 如果这是一个非事务性更改，获取我们预计要使用的快照。
	 * 我们只有在快照一致且此更改不应被跳过时才能到达这里。
	 *
	 * 对于事务性更改，我们不需要快照，我们将使用
	 * 重排序缓冲区维护的常规快照。我们只是将其保留为 NULL。
	 */
	if (!fc_message->transactional)
		fc_snapshot = SnapBuildGetOrBuildSnapshot(fc_builder, fc_xid);

	ReorderBufferQueueMessage(fc_ctx->reorder, fc_xid, fc_snapshot, fc_buf->endptr,
							  fc_message->transactional,
							  fc_message->message, /* 消息的第一部分是
												 * 前缀 */
							  fc_message->message_size,
							  fc_message->message + fc_message->prefix_size);
}

/*
 * 在不同类型的提交记录之间整合提交记录处理。
 *
 * 'two_phase' 表示调用者希望在两个阶段处理事务，首先处理
 * prepare（如果尚未完成），然后处理 commit_prepared。
 */
static void fc_DecodeCommit(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf,
			 xl_xact_parsed_commit *fc_parsed, TransactionId fc_xid,
			 bool fc_two_phase)
{
	XLogRecPtr	fc_origin_lsn = InvalidXLogRecPtr;
	TimestampTz fc_commit_time = fc_parsed->xact_time;
	RepOriginId fc_origin_id = XLogRecGetOrigin(fc_buf->record);
	int			fc_i;

	if (fc_parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
	{
		fc_origin_lsn = fc_parsed->origin_lsn;
		fc_commit_time = fc_parsed->origin_timestamp;
	}
	
#ifdef FDD
	//查看xid是否正在运行
	//FDB_WaitForXidProcEndTransaction(xid);
	//要判断是否是failover slot
	if (fc_ctx->failover_logical)
	{
		//检查对应的LSN是否在备机已经处理完成
		FDB_WaitForLsnEndTransaction(fc_buf->endptr, (void*)fc_ctx);
	}
#endif
	/*
	 * 如果 COMMIT 记录包含无效消息，则可能有目录更改。
	 * 可能我们在从提交记录开始解码时没有将此事务标记为
	 * 包含目录更改，而没有解码事务的其他更改。因此，我们确保
	 * 将此类事务标记为包含目录更改。
	 *
	 * 这必须在 SnapBuildCommitTxn() 之前完成，以便我们可以将
	 * 这些事务包含在历史快照中。
	 */
	if (fc_parsed->xinfo & XACT_XINFO_HAS_INVALS)
		SnapBuildXidSetCatalogChanges(fc_ctx->snapshot_builder, fc_xid,
									  fc_parsed->nsubxacts, fc_parsed->subxacts,
									  fc_buf->origptr);

	SnapBuildCommitTxn(fc_ctx->snapshot_builder, fc_buf->origptr, fc_xid,
					   fc_parsed->nsubxacts, fc_parsed->subxacts);

	/* ----
	 * 检查我们是否对该特定事务感兴趣，并告诉重排序缓冲区
	 * 如果不感兴趣则忘记（子）事务的内容。
	 *
	 * 我们不能仅在这里使用 ReorderBufferAbort()，因为我们需要
	 * 执行事务的无效操作。如果我们只是跳过该事务，当前不需要
	 * 这样做，因为我们目前只在启动时这样做，以获取客户端
	 * 需要的第一个事务。由于在开始读取 WAL 之前我们已重置
	 * 了目录缓存，并且我们尚未接触任何目录，因此不可能有任何
	 * 无效的东西。但是如果我们因为该提交发生在另一个数据库
	 * 而“忘记”它，则无效操作可能很重要，因为它们可能是
	 * 针对共享目录的，而我们可能已经将数据加载到相关的
	 * syscaches 中。
	 * ---
	 */
#ifdef FDDQ
	if (fc_DecodeTXNNeedSkip(fc_ctx, fc_buf, fc_parsed->dbId, fc_origin_id, fc_parsed->xinfo))
#else
	if (fc_DecodeTXNNeedSkip(fc_ctx, fc_buf, fc_parsed->dbId, fc_origin_id))
#endif
	{
		for (fc_i = 0; fc_i < fc_parsed->nsubxacts; fc_i++)
		{
			ReorderBufferForget(fc_ctx->reorder, fc_parsed->subxacts[fc_i], fc_buf->origptr);
		}
		ReorderBufferForget(fc_ctx->reorder, fc_xid, fc_buf->origptr);

		return;
	}

	/* 告诉重排序缓冲区关于存活的子事务 */
	for (fc_i = 0; fc_i < fc_parsed->nsubxacts; fc_i++)
	{
		ReorderBufferCommitChild(fc_ctx->reorder, fc_xid, fc_parsed->subxacts[fc_i],
								 fc_buf->origptr, fc_buf->endptr);
	}

	/*
	 * 如果事务数据已解码，则发送最终提交记录，否则，处理
	 * 整个事务。
	 */
	if (fc_two_phase)
	{
		ReorderBufferFinishPrepared(fc_ctx->reorder, fc_xid, fc_buf->origptr, fc_buf->endptr,
									SnapBuildGetTwoPhaseAt(fc_ctx->snapshot_builder),
									fc_commit_time, fc_origin_id, fc_origin_lsn,
									fc_parsed->twophase_gid, true);
	}
	else
	{
		ReorderBufferCommit(fc_ctx->reorder, fc_xid, fc_buf->origptr, fc_buf->endptr,
							fc_commit_time, fc_origin_id, fc_origin_lsn);
	}

	/*
	 * 在事务准备/提交/中止时更新解码统计信息。
	 * 此外，当我们溢出或流式传输更改时，我们会发送统计信息
	 * 以避免在解码中断的情况下丢失它们。尚不清楚
	 * 发送的频率比这更频繁或更不频繁是否会更好。
	 */
	UpdateDecodingStats(fc_ctx);
}

/*
 * 解码 PREPARE 记录。与 DecodeCommit 中的逻辑类似。
 *
 * 注意，即使检测到并发中止，我们也不会跳过 prepare，
 * 因为在我们检测到中止之前我们可能已经发送了一些更改，
 * 在这种情况下我们需要在订阅者中中止这些更改。
 * 为了中止这些更改，我们发送 prepare，然后发送 rollback prepared，
 * 这在发布者端也发生过。现在，我们可以构造一个新的
 * 中止 API，在这种情况下，我们发送中止并跳过发送
 * prepared 和 rollback prepared，但这并不是那么简单
 * 因为在那时我们可能已经流式传输了该事务，在这种情况下
 * 它会在遇到回滚时处理。优化并发中止情况下并不是不可能，
 * 但它可能会引入与处理不同情况相关的设计复杂性，因此
 * 暂时将其搁置，因为这似乎不值得。
 */
static void fc_DecodePrepare(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf,
			  xl_xact_parsed_prepare *fc_parsed)
{
	SnapBuild  *fc_builder = fc_ctx->snapshot_builder;
	XLogRecPtr	fc_origin_lsn = fc_parsed->origin_lsn;
	TimestampTz fc_prepare_time = fc_parsed->xact_time;
	RepOriginId	fc_origin_id = XLogRecGetOrigin(fc_buf->record);
	int			fc_i;
	TransactionId fc_xid = fc_parsed->twophase_xid;

	if (fc_parsed->origin_timestamp != 0)
		fc_prepare_time = fc_parsed->origin_timestamp;

	/*
	 * 记住 txn 的准备信息，以便在需要时可以在
	 * commit prepared 中使用。参见 ReorderBufferFinishPrepared。
	 */
	if (!ReorderBufferRememberPrepareInfo(fc_ctx->reorder, fc_xid, fc_buf->origptr,
										  fc_buf->endptr, fc_prepare_time, fc_origin_id,
										  fc_origin_lsn))
		return;

	/* 如果未达到一致状态，则我们无法启动流式传输。 */
	if (SnapBuildCurrentState(fc_builder) < SNAPBUILD_CONSISTENT)
	{
		ReorderBufferSkipPrepare(fc_ctx->reorder, fc_xid);
		return;
	}

	/*
	 * 检查我们是否需要处理此事务。参见
	 * DecodeTXNNeedSkip 了解我们有时想要跳过事务的原因。
	 *
	 * 我们不能像在 DecodeCommit 中那样调用 ReorderBufferForget，
	 * 因为 txn 尚未提交，在提交之前移除此 txn 可能会导致
	 * 计算不正确的 restart_lsn。参见 SnapBuildProcessRunningXacts。
	 * 但是如果有无效操作，我们需要处理缓存无效操作，
	 * 原因见 DecodeCommit。
	 */
#ifdef FDDQ
	if (fc_DecodeTXNNeedSkip(fc_ctx, fc_buf, fc_parsed->dbId, fc_origin_id,fc_parsed->xinfo))
#else
	if (fc_DecodeTXNNeedSkip(fc_ctx, fc_buf, fc_parsed->dbId, fc_origin_id))
#endif
	{
		ReorderBufferSkipPrepare(fc_ctx->reorder, fc_xid);
		ReorderBufferInvalidate(fc_ctx->reorder, fc_xid, fc_buf->origptr);
		return;
	}

	/* 告诉重排序缓冲区关于存活的子事务。 */
	for (fc_i = 0; fc_i < fc_parsed->nsubxacts; fc_i++)
	{
		ReorderBufferCommitChild(fc_ctx->reorder, fc_xid, fc_parsed->subxacts[fc_i],
								 fc_buf->origptr, fc_buf->endptr);
	}

	/* 按顺序重放所有事务 + 子事务的操作 */
	ReorderBufferPrepare(fc_ctx->reorder, fc_xid, fc_parsed->twophase_gid);

	/*
	 * 在事务准备/提交/中止时更新解码统计信息。
	 * 此外，当我们溢出或流式传输更改时，我们会发送统计信息
	 * 以避免在解码中断的情况下丢失它们。尚不清楚
	 * 发送的频率比这更频繁或更不频繁是否会更好。
	 */
	UpdateDecodingStats(fc_ctx);
}


/*
 * 从各种形式的中止记录中获取数据并将其传递到
 * snapbuild.c 和 reorderbuffer.c。
 *
 * 'two_phase' 表示完成已准备好的事务。
 */
static void fc_DecodeAbort(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf,
			xl_xact_parsed_abort *fc_parsed, TransactionId fc_xid,
			bool fc_two_phase)
{
	int			fc_i;
	XLogRecPtr	fc_origin_lsn = InvalidXLogRecPtr;
	TimestampTz fc_abort_time = fc_parsed->xact_time;
	RepOriginId	fc_origin_id = XLogRecGetOrigin(fc_buf->record);
	bool		fc_skip_xact;

	if (fc_parsed->xinfo & XACT_XINFO_HAS_ORIGIN)
	{
		fc_origin_lsn = fc_parsed->origin_lsn;
		fc_abort_time = fc_parsed->origin_timestamp;
	}

	
/*
	 * 检查我们是否需要处理此事务。请参见
	 * DecodeTXNNeedSkip，了解我们为什么有时想跳过该事务的原因。
	 */
#ifdef FDDQ
	fc_skip_xact = fc_DecodeTXNNeedSkip(fc_ctx, fc_buf, fc_parsed->dbId, fc_origin_id, fc_parsed->xinfo);
#else
	fc_skip_xact = fc_DecodeTXNNeedSkip(fc_ctx, fc_buf, fc_parsed->dbId, fc_origin_id);
#endif
	/*
	 * 发送准备事务的最终回滚记录，除非我们需要跳过它。对于非两阶段事务，简单地忘记该事务。
	 */
	if (fc_two_phase && !fc_skip_xact)
	{
		ReorderBufferFinishPrepared(fc_ctx->reorder, fc_xid, fc_buf->origptr, fc_buf->endptr,
									InvalidXLogRecPtr,
									fc_abort_time, fc_origin_id, fc_origin_lsn,
									fc_parsed->twophase_gid, false);
	}
	else
	{
		for (fc_i = 0; fc_i < fc_parsed->nsubxacts; fc_i++)
		{
			ReorderBufferAbort(fc_ctx->reorder, fc_parsed->subxacts[fc_i],
							   fc_buf->record->EndRecPtr);
		}

		ReorderBufferAbort(fc_ctx->reorder, fc_xid, fc_buf->record->EndRecPtr);
	}

	/* 更新解码统计信息 */
	UpdateDecodingStats(fc_ctx);
}

/*
 * 解析 XLOG_HEAP_INSERT（而不是 MULTI_INSERT！）记录到 tuplebufs。
 *
 * 删除操作可以包含新元组。
 */
static void fc_DecodeInsert(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf)
{
	Size		fc_datalen;
	char	   *fc_tupledata;
	Size		fc_tuplelen;
	XLogReaderState *fc_r = fc_buf->record;
	xl_heap_insert *fc_xlrec;
	ReorderBufferChange *fc_change;
	RelFileNode fc_target_node;

	fc_xlrec = (xl_heap_insert *) XLogRecGetData(fc_r);

	/*
	 * 忽略没有新元组的插入记录（这确实发生在
	 * raw_heap_insert 将 TOAST 记录标记为 HEAP_INSERT_NO_LOGICAL 时）。
	 */
	if (!(fc_xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE))
		return;

	/* 只对我们的数据库感兴趣 */
	XLogRecGetBlockTag(fc_r, 0, &fc_target_node, NULL, NULL);
	if (fc_target_node.dbNode != fc_ctx->slot->data.database)
		return;

	/* 输出插件不查找此来源，无需排队 */
	if (fc_FilterByOrigin(fc_ctx, XLogRecGetOrigin(fc_r)))
		return;

	fc_change = ReorderBufferGetChange(fc_ctx->reorder);
	if (!(fc_xlrec->flags & XLH_INSERT_IS_SPECULATIVE))
		fc_change->action = REORDER_BUFFER_CHANGE_INSERT;
	else
		fc_change->action = REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT;
	fc_change->origin_id = XLogRecGetOrigin(fc_r);

	memcpy(&fc_change->data.tp.relnode, &fc_target_node, sizeof(RelFileNode));

	fc_tupledata = XLogRecGetBlockData(fc_r, 0, &fc_datalen);
	fc_tuplelen = fc_datalen - SizeOfHeapHeader;

	fc_change->data.tp.newtuple =
		ReorderBufferGetTupleBuf(fc_ctx->reorder, fc_tuplelen);

	fc_DecodeXLogTuple(fc_tupledata, fc_datalen, fc_change->data.tp.newtuple);

	fc_change->data.tp.clear_toast_afterwards = true;

	ReorderBufferQueueChange(fc_ctx->reorder, XLogRecGetXid(fc_r), fc_buf->origptr,
							 fc_change,
							 fc_xlrec->flags & XLH_INSERT_ON_TOAST_RELATION);
}

/*
 * 解析 XLOG_HEAP_UPDATE 和 XLOG_HEAP_HOT_UPDATE，这些记录的布局相同，
 * 从 wal 转换为适当的 tuplebufs。
 *
 * 更新可能包含新元组和旧主键。
 */
static void fc_DecodeUpdate(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf)
{
	XLogReaderState *fc_r = fc_buf->record;
	xl_heap_update *fc_xlrec;
	ReorderBufferChange *fc_change;
	char	   *fc_data;
	RelFileNode fc_target_node;

	fc_xlrec = (xl_heap_update *) XLogRecGetData(fc_r);

	/* 只对我们的数据库感兴趣 */
	XLogRecGetBlockTag(fc_r, 0, &fc_target_node, NULL, NULL);
	if (fc_target_node.dbNode != fc_ctx->slot->data.database)
		return;

	/* 输出插件不查找此来源，无需排队 */
	if (fc_FilterByOrigin(fc_ctx, XLogRecGetOrigin(fc_r)))
		return;

	fc_change = ReorderBufferGetChange(fc_ctx->reorder);
	fc_change->action = REORDER_BUFFER_CHANGE_UPDATE;
	fc_change->origin_id = XLogRecGetOrigin(fc_r);
	memcpy(&fc_change->data.tp.relnode, &fc_target_node, sizeof(RelFileNode));

	if (fc_xlrec->flags & XLH_UPDATE_CONTAINS_NEW_TUPLE)
	{
		Size		fc_datalen;
		Size		fc_tuplelen;

		fc_data = XLogRecGetBlockData(fc_r, 0, &fc_datalen);

		fc_tuplelen = fc_datalen - SizeOfHeapHeader;

		fc_change->data.tp.newtuple =
			ReorderBufferGetTupleBuf(fc_ctx->reorder, fc_tuplelen);

		fc_DecodeXLogTuple(fc_data, fc_datalen, fc_change->data.tp.newtuple);
	}

	if (fc_xlrec->flags & XLH_UPDATE_CONTAINS_OLD)
	{
		Size		fc_datalen;
		Size		fc_tuplelen;

		/* 注意，记录中的剩余数据未对齐 */
		fc_data = XLogRecGetData(fc_r) + SizeOfHeapUpdate;
		fc_datalen = XLogRecGetDataLen(fc_r) - SizeOfHeapUpdate;
		fc_tuplelen = fc_datalen - SizeOfHeapHeader;

		fc_change->data.tp.oldtuple =
			ReorderBufferGetTupleBuf(fc_ctx->reorder, fc_tuplelen);

		fc_DecodeXLogTuple(fc_data, fc_datalen, fc_change->data.tp.oldtuple);
	}

	fc_change->data.tp.clear_toast_afterwards = true;

	ReorderBufferQueueChange(fc_ctx->reorder, XLogRecGetXid(fc_r), fc_buf->origptr,
							 fc_change, false);
}

/*
 * 将 XLOG_HEAP_DELETE 从 wal 解析为适当的 tuplebufs。
 *
 * 删除操作可能包含旧主键。
 */
static void fc_DecodeDelete(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf)
{
	XLogReaderState *fc_r = fc_buf->record;
	xl_heap_delete *fc_xlrec;
	ReorderBufferChange *fc_change;
	RelFileNode fc_target_node;

	fc_xlrec = (xl_heap_delete *) XLogRecGetData(fc_r);

	/* 只对我们的数据库感兴趣 */
	XLogRecGetBlockTag(fc_r, 0, &fc_target_node, NULL, NULL);
	if (fc_target_node.dbNode != fc_ctx->slot->data.database)
		return;

	/* 输出插件不查找此来源，无需排队 */
	if (fc_FilterByOrigin(fc_ctx, XLogRecGetOrigin(fc_r)))
		return;

	fc_change = ReorderBufferGetChange(fc_ctx->reorder);

	if (fc_xlrec->flags & XLH_DELETE_IS_SUPER)
		fc_change->action = REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT;
	else
		fc_change->action = REORDER_BUFFER_CHANGE_DELETE;

	fc_change->origin_id = XLogRecGetOrigin(fc_r);

	memcpy(&fc_change->data.tp.relnode, &fc_target_node, sizeof(RelFileNode));

	/* 存储旧主键 */
	if (fc_xlrec->flags & XLH_DELETE_CONTAINS_OLD)
	{
		Size		fc_datalen = XLogRecGetDataLen(fc_r) - SizeOfHeapDelete;
		Size		fc_tuplelen = fc_datalen - SizeOfHeapHeader;

		Assert(XLogRecGetDataLen(fc_r) > (SizeOfHeapDelete + SizeOfHeapHeader));

		fc_change->data.tp.oldtuple =
			ReorderBufferGetTupleBuf(fc_ctx->reorder, fc_tuplelen);

		fc_DecodeXLogTuple((char *) fc_xlrec + SizeOfHeapDelete,
						fc_datalen, fc_change->data.tp.oldtuple);
	}

	fc_change->data.tp.clear_toast_afterwards = true;

	ReorderBufferQueueChange(fc_ctx->reorder, XLogRecGetXid(fc_r), fc_buf->origptr,
							 fc_change, false);
}

/*
 * 从 wal 解析 XLOG_HEAP_TRUNCATE
 */
static void fc_DecodeTruncate(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf)
{
	XLogReaderState *fc_r = fc_buf->record;
	xl_heap_truncate *fc_xlrec;
	ReorderBufferChange *fc_change;

	fc_xlrec = (xl_heap_truncate *) XLogRecGetData(fc_r);

	/* 只对我们的数据库感兴趣 */
	if (fc_xlrec->dbId != fc_ctx->slot->data.database)
		return;

	/* 输出插件不查找此来源，无需排队 */
	if (fc_FilterByOrigin(fc_ctx, XLogRecGetOrigin(fc_r)))
		return;

	fc_change = ReorderBufferGetChange(fc_ctx->reorder);
	fc_change->action = REORDER_BUFFER_CHANGE_TRUNCATE;
	fc_change->origin_id = XLogRecGetOrigin(fc_r);
	if (fc_xlrec->flags & XLH_TRUNCATE_CASCADE)
		fc_change->data.truncate.cascade = true;
	if (fc_xlrec->flags & XLH_TRUNCATE_RESTART_SEQS)
		fc_change->data.truncate.restart_seqs = true;
	fc_change->data.truncate.nrelids = fc_xlrec->nrelids;
	fc_change->data.truncate.relids = ReorderBufferGetRelids(fc_ctx->reorder,
														  fc_xlrec->nrelids);
	memcpy(fc_change->data.truncate.relids, fc_xlrec->relids,
		   fc_xlrec->nrelids * sizeof(Oid));
	ReorderBufferQueueChange(fc_ctx->reorder, XLogRecGetXid(fc_r),
							 fc_buf->origptr, fc_change, false);
}

/*
 * 将 XLOG_HEAP2_MULTI_INSERT_insert 记录解码为多个 tuplebufs。
 *
 * 目前 MULTI_INSERT 将始终包含完整的元组。
 */
static void fc_DecodeMultiInsert(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf)
{
	XLogReaderState *fc_r = fc_buf->record;
	xl_heap_multi_insert *fc_xlrec;
	int			fc_i;
	char	   *fc_data;
	char	   *fc_tupledata;
	Size		fc_tuplelen;
	RelFileNode fc_rnode;

	fc_xlrec = (xl_heap_multi_insert *) XLogRecGetData(fc_r);

	/*
	 * 忽略没有新元组的插入记录。这发生在对目录或非持久关系进行
	 * multi_insert 时。
	 */
	if (!(fc_xlrec->flags & XLH_INSERT_CONTAINS_NEW_TUPLE))
		return;

	/* 只对我们的数据库感兴趣 */
	XLogRecGetBlockTag(fc_r, 0, &fc_rnode, NULL, NULL);
	if (fc_rnode.dbNode != fc_ctx->slot->data.database)
		return;

	/* 输出插件不查找此来源，无需排队 */
	if (fc_FilterByOrigin(fc_ctx, XLogRecGetOrigin(fc_r)))
		return;

	/*
	 * 我们知道这个 multi_insert 不是用于目录的，因此这个块
	 * 即使进行完整页面写入也应该始终有数据。
	 */
	fc_tupledata = XLogRecGetBlockData(fc_r, 0, &fc_tuplelen);
	Assert(fc_tupledata != NULL);

	fc_data = fc_tupledata;
	for (fc_i = 0; fc_i < fc_xlrec->ntuples; fc_i++)
	{
		ReorderBufferChange *fc_change;
		xl_multi_insert_tuple *fc_xlhdr;
		int			fc_datalen;
		ReorderBufferTupleBuf *fc_tuple;
		HeapTupleHeader fc_header;

		fc_change = ReorderBufferGetChange(fc_ctx->reorder);
		fc_change->action = REORDER_BUFFER_CHANGE_INSERT;
		fc_change->origin_id = XLogRecGetOrigin(fc_r);

		memcpy(&fc_change->data.tp.relnode, &fc_rnode, sizeof(RelFileNode));

		fc_xlhdr = (xl_multi_insert_tuple *) SHORTALIGN(fc_data);
		fc_data = ((char *) fc_xlhdr) + SizeOfMultiInsertTuple;
		fc_datalen = fc_xlhdr->datalen;

		fc_change->data.tp.newtuple =
			ReorderBufferGetTupleBuf(fc_ctx->reorder, fc_datalen);

		fc_tuple = fc_change->data.tp.newtuple;
		fc_header = fc_tuple->tuple.t_data;

		/* 不是基于磁盘的元组 */
		ItemPointerSetInvalid(&fc_tuple->tuple.t_self);

		/*
		 * 这只有在重新组装事务后才能确定。
		 */
		fc_tuple->tuple.t_tableOid = InvalidOid;

		fc_tuple->tuple.t_len = fc_datalen + SizeofHeapTupleHeader;

		memset(fc_header, 0, SizeofHeapTupleHeader);

		memcpy((char *) fc_tuple->tuple.t_data + SizeofHeapTupleHeader,
			   (char *) fc_data,
			   fc_datalen);
		fc_header->t_infomask = fc_xlhdr->t_infomask;
		fc_header->t_infomask2 = fc_xlhdr->t_infomask2;
		fc_header->t_hoff = fc_xlhdr->t_hoff;

		/*
		 * 仅在由一次 heap_multi_insert() 调用发出的最后一个
		 * xl_multi_insert_tuple 记录中的最后一行之后重置 toast 重新组装状态。
		 */
		if (fc_xlrec->flags & XLH_INSERT_LAST_IN_MULTI &&
			(fc_i + 1) == fc_xlrec->ntuples)
			fc_change->data.tp.clear_toast_afterwards = true;
		else
			fc_change->data.tp.clear_toast_afterwards = false;

		ReorderBufferQueueChange(fc_ctx->reorder, XLogRecGetXid(fc_r),
								 fc_buf->origptr, fc_change, false);

		/* 移动到下一个 xl_multi_insert_tuple 条目 */
		fc_data += fc_datalen;
	}
	Assert(fc_data == fc_tupledata + fc_tuplelen);
}

/*
 * 从 wal 中解析 XLOG_HEAP_CONFIRM 以进行确认更改。
 *
 * 这非常简单，所有的状态本质上已经通过
 * 投机插入设置好了。
 */
static void fc_DecodeSpecConfirm(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf)
{
	XLogReaderState *fc_r = fc_buf->record;
	ReorderBufferChange *fc_change;
	RelFileNode fc_target_node;

	/* 只对我们的数据库感兴趣 */
	XLogRecGetBlockTag(fc_r, 0, &fc_target_node, NULL, NULL);
	if (fc_target_node.dbNode != fc_ctx->slot->data.database)
		return;

	/* 输出插件不查找此来源，无需排队 */
	if (fc_FilterByOrigin(fc_ctx, XLogRecGetOrigin(fc_r)))
		return;

	fc_change = ReorderBufferGetChange(fc_ctx->reorder);
	fc_change->action = REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM;
	fc_change->origin_id = XLogRecGetOrigin(fc_r);

	memcpy(&fc_change->data.tp.relnode, &fc_target_node, sizeof(RelFileNode));

	fc_change->data.tp.clear_toast_afterwards = true;

	ReorderBufferQueueChange(fc_ctx->reorder, XLogRecGetXid(fc_r), fc_buf->origptr,
							 fc_change, false);
}


/*
 * 将由 heap_insert、heap_update 和 heap_delete（但不包括 heap_multi_insert）记录的 HeapTuple
 * 读取到 tuplebuf 中。
 *
 * 记录中的大小 'len' 和指针 'data' 需要
 * 在外部计算，因为它们是特定于记录的。
 */
static void fc_DecodeXLogTuple(char *fc_data, Size fc_len, ReorderBufferTupleBuf *fc_tuple)
{
	xl_heap_header fc_xlhdr;
	int			fc_datalen = fc_len - SizeOfHeapHeader;
	HeapTupleHeader fc_header;

	Assert(fc_datalen >= 0);

	fc_tuple->tuple.t_len = fc_datalen + SizeofHeapTupleHeader;
	fc_header = fc_tuple->tuple.t_data;

	/* 不是基于磁盘的元组 */
	ItemPointerSetInvalid(&fc_tuple->tuple.t_self);

	/* 我们只能在重新组装事务后弄清楚这一点 */
	fc_tuple->tuple.t_tableOid = InvalidOid;

	/* 数据未存储对齐，复制到对齐存储中 */
	memcpy((char *) &fc_xlhdr,
		   fc_data,
		   SizeOfHeapHeader);

	memset(fc_header, 0, SizeofHeapTupleHeader);

	memcpy(((char *) fc_tuple->tuple.t_data) + SizeofHeapTupleHeader,
		   fc_data + SizeOfHeapHeader,
		   fc_datalen);

	fc_header->t_infomask = fc_xlhdr.t_infomask;
	fc_header->t_infomask2 = fc_xlhdr.t_infomask2;
	fc_header->t_hoff = fc_xlhdr.t_hoff;
}

/*
 * 检查我们是否对这个特定事务感兴趣。
 *
 * 我们可能对这个事务不感兴趣的原因有几个：
 * 1）我们可能对解码该 LSN 之前的事务不感兴趣。这可能是因为我们之前已解码，并且现在只是
 *	  重新启动，或者如果我们尚未组装一致的快照。
 * 2）事务发生在另一个数据库中。
 * 3）输出插件对源不感兴趣。
 * 4）我们正在进行快进。
 */
static bool fc_DecodeTXNNeedSkip(LogicalDecodingContext *fc_ctx, XLogRecordBuffer *fc_buf,
#ifdef FDDQ
				  Oid fc_txn_dbid, RepOriginId fc_origin_id,uint32 fc_xinfo)
#else
				  Oid fc_txn_dbid, RepOriginId fc_origin_id)
#endif
{
	return (SnapBuildXactNeedsSkip(fc_ctx->snapshot_builder, fc_buf->origptr) ||
			(fc_txn_dbid != InvalidOid && fc_txn_dbid != fc_ctx->slot->data.database) ||
#ifdef FDDQ
		(fc_xinfo & XACT_XINFO_LOGIREPL_OFF) ||
#endif
			fc_ctx->fast_forward || fc_FilterByOrigin(fc_ctx, fc_origin_id));
}
